package com.atguigu.gmall.realtime.utils

import java.util

import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010.OffsetRange
import redis.clients.jedis.Jedis

import scala.collection.mutable

object MyOffsetUtils {
  /**
   * 将offset信息存储到redis中
   *
   * offset结构:  (GTP)group + topic + partition => offset
   *                   g : t      0 => 100
   *                   g : t      1 => 101
   *                          =>
   *                   g : t      2 => 99
   *                   g : t      3 => 100
   * Redis中的存储:
   * type:    hash
   * key:     offset:[group]:[topic]
   * value:   [partition => offset] ...
   * 写入API:  hset
   * 读取API:  hgetall
   * 是否过期: 不过期
   */
  def SaveOffset(topic: String, groupId: String,ranges: Array[OffsetRange]): Unit = {
    if(ranges != null && ranges.size > 0){
      val jedis: Jedis = MyRedisUtils.get()
      val hashMap: util.HashMap[String, String] = new util.HashMap[String,String]()
      val key = s"offset:$topic:$groupId"
      for (offset <- ranges) {
        val partition: Int = offset.partition
        val endOffset: Long = offset.untilOffset
        hashMap.put(partition.toString,endOffset.toString)
      }
      println("保存offset:"+ hashMap)
      jedis.hset(key,hashMap)
      jedis.close()
    }
  }

  /**
   * 从redis中读取offset信息
   * Map[TopicPartition,Long] :
   * TopicPartition : 封装topic 和 partition
   * Long:  就是offset
   */
  def ReadOffset(topic: String, groupId: String): Map[TopicPartition, Long] = {
    val offsetMap: mutable.Map[TopicPartition, Long] = mutable.Map[TopicPartition, Long]()
    val jedis: Jedis = MyRedisUtils.get()
    val keys = s"offset:$topic:$groupId"
    //从redis中取出数据
    val offsetData: util.Map[String, String] = jedis.hgetAll(keys)
    println("读到offset: " + offsetData)
    if (offsetData != null && offsetData.size() > 0) {
      //将java集合转换成scala集合 需要调用该隐式转换
      import scala.collection.JavaConverters._
      for ((partition, offset) <- offsetData.asScala) {
        val topicP: TopicPartition = new TopicPartition(topic, partition.toInt)
        offsetMap.put(topicP, offset.toLong)
      }
    }
    MyRedisUtils.close(jedis)
    //由于需要的是不可变map所以需要将可变转为不可变
    offsetMap.toMap
  }
}
